package com.test.cn.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}

case class LrisVo(var Id: String, var SepalLengthCm: Double, var SepalWidthCm: Double, var PetalLengthCm: Double, var PetalWidthCm: Double, var Species: String)

object KNNQuestion {

  def main(args: Array[String]): Unit = {
    //
    val conf: SparkConf = new SparkConf()
      .setAppName(this.getClass.getCanonicalName.init)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val K = 10
    //读取文件
    val textRDDA: RDD[String] = sc.textFile("spark_scala_home_work/src/data/Iris_knn_a.csv")
    val lrisRDD_A: RDD[LrisVo] = textRDDA.map { case str => {
      val strs: Array[String] = str.split(",")
      LrisVo(strs(0), strs(1).toDouble, strs(2).toDouble, strs(3).toDouble, strs(4).toDouble, strs(5))
    }
    }


    val textRDDB: RDD[String] = sc.textFile("spark_scala_home_work/src/data/Iris_knn_b.csv")
    val lrisRDD_B: RDD[LrisVo] = textRDDB.map { case str => {
      val strs: Array[String] = str.split(",")
      LrisVo(strs(0), strs(1).toDouble, strs(2).toDouble, strs(3).toDouble, strs(4).toDouble, strs(5))
    }
    }


    def getInstance(v1:LrisVo,v2:LrisVo):Double={
      Math.sqrt(  Math.pow((v1.PetalLengthCm - v2.PetalLengthCm),2)
        + Math.pow(v1.PetalWidthCm -v2.PetalWidthCm,2)
        + Math.pow(v1.SepalLengthCm -v2.SepalLengthCm,2)
        + Math.pow(v1.SepalWidthCm -v2.SepalWidthCm,2))
    }

    //求数据集Y中每个点，到数据集X到每个点的距离D
    val value: RDD[(String, (String, Double))] = lrisRDD_B.cartesian(lrisRDD_A).map { case (v1: LrisVo, v2: LrisVo) => {
      (v1.Id, (v2.Species, getInstance(v1, v2)))
    }
    }



    //找到数据集D中最小K个点
    val kvalue: RDD[(String, List[(String, Double)])] = value.groupByKey().mapValues(_.toList.sortWith(_._2 < _._2).take(K))


//    val tuples: Array[(String, List[(String, Double)])] = kvalue.collect()
//
//    for (elem <- tuples) {
//      println(elem)
//    }

    //求K个点的分类情况,拿到得分最多的分类
    val valueB: RDD[(String, (String, Int))] = kvalue.mapValues { case list: List[(String, Double)] => {
      val stringToTuples: Map[String, List[(String, Double)]] = list.groupBy(_._1)
      val stringToInt: Map[String, Int] = list.groupBy(_._1).mapValues(_.size)
      val tuple: (String, Int) = stringToInt.maxBy(_._2)
      tuple
    }
    }
    val tuples: Array[(String, (String, Int))] = valueB.collect()
    for (elem <- tuples) {
      println(elem)
    }




    sc.stop()
  }


}
